package com.uptown.matrix.core.consumer.rocketmq;

import com.uptown.matrix.core.annotation.MatrixMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

/**
 * @author: uptown
 * @date: 2022/4/13 23:05
 */

@Slf4j
@Component
public class RocketMqConsumerHandler implements DisposableBean {

    @Resource
    Map<String, DefaultMQPushConsumer> rocketMqConsumerMap;

    public void createSubscription(Object bean, MatrixMessageListener annotation, Method method, String groupTopicKey) throws Throwable {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(annotation.consumerGroup());
        consumer.setNamesrvAddr(annotation.bootServer());
        rocketMqConsumerMap.put(groupTopicKey, consumer);
        consumer.subscribe(annotation.streamKey(),annotation.tag());
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            try {
                method.invoke(bean, list);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            } catch (Exception e) {

                log.error(e.getMessage(), e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });
        consumer.start();
    }


    @Override
    public void destroy() {
        rocketMqConsumerMap.forEach((topic, consumer) -> {
            consumer.shutdown();
        });
    }
}
